package org.cliff.message.platform.core.resend;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 消息重发执行器
 * 
 * @author 刘文
 * @date 2023/10/25
 */
@Slf4j
public class ResendMessageExecutor {

    /**
     * 接收消息重发执行器
     */
    public static ResendMessageExecutor RECEIVED_MESSAGE_RESEND_EXECUTOR =
        new ResendMessageExecutor("ReceivedMessageResend");

    /**
     * 重发最大并发任务数
     */
    public final int RESEND_MAX_CONCURRENT_TASKS = 5;

    private final AtomicInteger rejectedTaskCount = new AtomicInteger(0);

    /**
     * 接收消息重发线程池
     */
    private final ThreadPoolExecutor threadPoolExecutor;

    /**
     * 重发送并发控制锁
     */
    private final Semaphore resendMessageSemaphore = new Semaphore(RESEND_MAX_CONCURRENT_TASKS);

    public ResendMessageExecutor(String namePrefix) {
        // 0核心线程池，便于大量任务的时候，能快速增大线程。
        threadPoolExecutor = new ThreadPoolExecutor(0, RESEND_MAX_CONCURRENT_TASKS, 3, TimeUnit.SECONDS,
            // 阻塞队列为并发数-1是为了能及时生成新的线程，否则5个阻塞队列不会生成新的线程。
            new LinkedBlockingDeque<>(RESEND_MAX_CONCURRENT_TASKS - 1),
            ThreadFactoryBuilder.create().setNamePrefix(namePrefix).build(), new MessagePlatRejectedHandler());
    }

    /**
     * 运行一批次任务
     * 
     * @param tasks 待执行的任务
     * @param stopSubmitTime 停止提交时间,过了这个时间不要提交任务,避免任务重复执行.
     */
    public int executeTasks(List<? extends Runnable> tasks, long stopSubmitTime) {
        int submittedTasks = 0;
        for (Runnable task : tasks) {
            // 获取任务信号量，避免过度添加任务
            try {
                while (!resendMessageSemaphore.tryAcquire(500, TimeUnit.MILLISECONDS)) {
                    continue;
                }
            } catch (InterruptedException e) {
                log.error("Interrupted while getting lock.", e);
            }

            // 超时控制
            if (System.currentTimeMillis() >= stopSubmitTime) {
                log.debug("Stop submitting task for time exceed");
                break;
            }

            threadPoolExecutor.submit(() -> {
                try {
                    task.run();
                } finally {
                    // 任务执行结束，提交
                    resendMessageSemaphore.release();
                }
            });
            submittedTasks++;
        }
        return submittedTasks;
    }

    public int getRejectedTaskCount() {
        return rejectedTaskCount.get();
    }

    /**
     * 消息平台任务拒绝处理器
     */
    private class MessagePlatRejectedHandler implements RejectedExecutionHandler {

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            resendMessageSemaphore.release();
            log.warn("One task is rejected, total rejected tasks {}", rejectedTaskCount.incrementAndGet());
        }
    }

}
